package kafkaTest;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class MyProducter {

	public static void main(String[] args) {
		//属性配置
		Properties properties=new Properties();
		//服务
		properties.put("bootstrap.servers","192.168.12.151:9092,192.168.12.152:9092,192.168.12.154:9092");
		//序列化
		properties.put("key.serializer","org.apache.kafka.common.serialization.IntegerSerializer");
		properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
		//构造生产者创建链接
		KafkaProducer<Integer, String>producter=new KafkaProducer<Integer, String>(properties);
		//主题
		String topic="test";
		int messageNo=1;
		try {
			while (true) {
				//构造消息
				ProducerRecord<Integer, String> record = new ProducerRecord<Integer, String>(topic, messageNo,
						"message_" + messageNo++);
				//发送消息
				producter.send(record);
				System.out.println("success....");
				Thread.sleep(1000);
			} 
		} catch (Exception e) {
           e.printStackTrace();
		}finally {
			producter.close();
		}
	}
}
